KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig #16458

brandboat · 2024-06-26T14:56:12Z

related to https://issues.apache.org/jira/browse/KAFKA-16909

as title, this pr follows RemoteLogManagerConfig.java, pass AbstractConfig to constructor.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712

@brandboat thanks for this patch

chia7712 · 2024-06-26T17:00:11Z

core/src/main/scala/kafka/server/BrokerServer.scala

-        config.consumerGroupMigrationPolicy,
-        config.offsetsTopicCompressionType
-      )
+      val groupCoordinatorConfig = new GroupCoordinatorConfig(config)


Could we add a method to KafkaConfig to return GroupCoordinatorConfig? With that change, we can remove all GroupCoordinatorConfig-related getters from KafkaConfig

Thank you for the suggestion! I have added the method to KafkaConfig to return GroupCoordinatorConfig and removed all related getters. This change is included in the latest commits.

…onType

chia7712

@brandboat thanks for this patch

chia7712 · 2024-06-30T14:41:38Z

...-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java

@@ -2576,15 +2577,15 @@ public void testCleanupExpiredOffsetsWithPendingTransactionalOffsets() {

        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder()
            .withGroupMetadataManager(groupMetadataManager)
-            .withOffsetsRetentionMs(1000)
+            .withOffsetsRetentionMinutes(1)


not sure why we need this change. Also, the value is changed from 1 second to 1 minute?

Actually the original time unit of config offsets.retention.minutes is in minutes, and before this refactor we pass offsets.retention in millis to GroupCoordinatorConfig constructor and use it in OffsetMetadataManagerTest, now we pass AbstractConfig (i.e. KafkaConfig) to GroupCoordinatorConfig, which means we should pass offsets.retention in minutes instead of ms

chia7712 · 2024-06-30T14:44:23Z

...coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java

+        configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.DISABLED.name());
+        configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_COMPRESSION_CODEC_CONFIG, (int) CompressionType.NONE.id);
+
+        return new GroupCoordinatorConfig(new GroupCoordinatorTestConfig(configs));


return new GroupCoordinatorConfig(new AbstractConfig(Utils.mergeConfigs(Arrays.asList( GroupCoordinatorConfig.GROUP_COORDINATOR_CONFIG_DEF, GroupCoordinatorConfig.NEW_GROUP_CONFIG_DEF, GroupCoordinatorConfig.OFFSET_MANAGEMENT_CONFIG_DEF, GroupCoordinatorConfig.CONSUMER_GROUP_CONFIG_DEF)), configs, false));

WDYT? we don't need the temporary class

Will do, thanks for the suggestion.

chia7712 · 2024-06-30T14:47:54Z

core/src/main/scala/kafka/server/KafkaConfig.scala

-  val groupCoordinatorAppendLingerMs = getInt(GroupCoordinatorConfig.GROUP_COORDINATOR_APPEND_LINGER_MS_CONFIG)
-
-  /** Consumer group configs */
-  val consumerGroupSessionTimeoutMs = getInt(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG)


@OmniaGM Could you please take a look? Does it follow your idea that we should move getters out of KafkaConfig?

Sorry for late response here, this looks great! I think the validators could also be moved out as well

Sure, we will file a MINOR for it

chia7712

LGTM

dajac

@brandboat @chia7712 Thanks for the PR. Overall, I agree with the change. However, I have left a few suggestions for consideration. Please let me know what you think.

dajac · 2024-07-02T08:57:43Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java

+    public ConsumerGroupMigrationPolicy consumerGroupMigrationPolicy() {
+        return ConsumerGroupMigrationPolicy.parse(
+                config.getString(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG));
+    }


I would like to point out that this change is a bit risky in my opinion because it does not ensure during the startup that the migration policy is really correct. If it fails somehow, it will fail later on when consumerGroupMigrationPolicy is accessed for the first time. I wonder if we should keep local attributes and initialize them in the constructor.

The validation is addressed by the GroupCoordinatorConfig's config definition.

https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaConfig.scala#L182

Hence, the string value is valid in constructing GroupCoordinatorConfig

Indeed. I was thinking about the case where the validation has a bug or is not good enough.

dajac · 2024-07-02T09:00:51Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java

+    public int offsetCommitTimeoutMs() {
+        return config.getInt(GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG);
+    }


This is an example of the config which is accessed extremely frequently (there are other like this too). I think that having attributes would be better as it avoid having to look it up in the config every time.

This is a good question!

The approach of this PR is to make sure GroupCoordinatorConfig can see the latest configs to avoid potential bugs (#16394). However, it has side effect which brings extra cost: "volatile" and "lookup/parse".

The cost of "volatile" (or other similar sync trick) is required if we make GroupCoordinatorConfig see latest configs.

The cost of "lookup/parse" could be eliminated if we do a bit refactor for it. For example, we pass Supplier<GroupCoordinatorConfig> instead of GroupCoordinatorConfig to the callers. By that changes, we can make GroupCoordinatorConfig have all immutable pre-created local attributes. The impl of Supplier<GroupCoordinatorConfig> will be generated by KafkaConfig and it looks like AtomicReference::get. However, the side effect is that the usage will get a little ugly: "config.numThreads()" -> "config.get().numThreads()"

WDYT? BTW, we had a related discussion in https://issues.apache.org/jira/browse/KAFKA-17001

None of the configs in this class are dynamic so I am not sure that it is worth it. We could perhaps mention it in the javadoc of the class. If we introduce a dynamic config, we should indeed not use an attribute for it.

For the context, in KafkaConfig, we always had the distinction between val (static values) and def (dynamic ones). We could do the same here, I suppose.

chia7712 · 2024-07-03T17:11:57Z

I double-check all configs and yes all of them are not dynamic. Maybe we don't need to be over-engineering for now. Hence, we can have a PR for following changes.

re-introduce attributes to GroupCoordinatorConfig @dajac comment
move GroupCoordinatorConfig validation from KafkaConfig to GroupCoordinatorConfig ( @OmniaGM comment)
do validation in construction of GroupCoordinatorConfig. I guess this is a bit different to @OmniaGM idea (KAFKA-15853: Refactor ShareGroupConfig with AbstractConfig #16506 add a method validate to ShareGroupConfig instead of validating configs in construction). I prefer to validate all configs in construction as KafakConfig do validation in construction too.
add docs to GroupCoordinatorConfig to explain why we add attributes

@brandboat @dajac @OmniaGM PTAL, I hope this can be a guideline for all similar config class

chia7712 · 2024-07-04T11:59:00Z

@brandboat I open https://issues.apache.org/jira/browse/KAFKA-17081 as follow-up. And it is assigned to you. PLEASE feel free to assign it back to me if you have no bandwidth

dajac · 2024-07-04T13:43:12Z

@chia7712 Thanks. Sounds good to me!

KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig

202a950

chia7712 reviewed Jun 26, 2024

View reviewed changes

brandboat added 3 commits June 29, 2024 20:45

Add method to return GroupCoordinatorConfig in KafkaConfig

a761752

Rename GroupCoordinatorConfig#compressionType to offsetTopicCompressi…

1231164

…onType

Merge remote-tracking branch 'origin/trunk' into KAFKA-16909

a23ea99

brandboat marked this pull request as draft June 29, 2024 13:48

brandboat added 2 commits June 29, 2024 21:56

Remove duplicate configs in GroupCoordinatorConfig

ffee5fc

Add more unit tests

524bd55

brandboat marked this pull request as ready for review June 29, 2024 14:05

brandboat added 2 commits June 29, 2024 22:08

Address comments

4c21247

Remove unsued import

bad7334

chia7712 reviewed Jun 30, 2024

View reviewed changes

brandboat added 2 commits June 30, 2024 23:21

Address comment

9e75d88

Merge remote-tracking branch 'origin/trunk' into KAFKA-16909

773effc

chia7712 approved these changes Jul 1, 2024

View reviewed changes

chia7712 merged commit 206d0f8 into apache:trunk Jul 1, 2024
1 check failed

brandboat deleted the KAFKA-16909 branch July 1, 2024 15:32

dajac reviewed Jul 2, 2024

View reviewed changes

chia7712 mentioned this pull request Jul 2, 2024

KAFKA-15853: Refactor ShareGroupConfig with AbstractConfig #16506

Open

3 tasks

brandboat mentioned this pull request Jul 4, 2024

KAFKA-17081 Tweak GroupCoordinatorConfig: re-introduce local attributes and validation #16524

Open

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig #16458

KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig #16458

brandboat commented Jun 26, 2024

chia7712 left a comment

chia7712 Jun 26, 2024

brandboat Jun 30, 2024

chia7712 left a comment

chia7712 Jun 30, 2024

brandboat Jun 30, 2024 •

edited

Loading

chia7712 Jun 30, 2024

brandboat Jun 30, 2024

chia7712 Jun 30, 2024

OmniaGM Jul 2, 2024

chia7712 Jul 2, 2024

chia7712 left a comment

dajac left a comment

dajac Jul 2, 2024

chia7712 Jul 2, 2024

dajac Jul 3, 2024

dajac Jul 2, 2024

chia7712 Jul 2, 2024

dajac Jul 3, 2024

chia7712 commented Jul 3, 2024

chia7712 commented Jul 4, 2024

dajac commented Jul 4, 2024

KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig #16458

KAFKA-16909 Refactor GroupCoordinatorConfig with AbstractConfig #16458

Conversation

brandboat commented Jun 26, 2024

Committer Checklist (excluded from commit message)

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brandboat Jun 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 commented Jul 3, 2024

chia7712 commented Jul 4, 2024

dajac commented Jul 4, 2024

brandboat Jun 30, 2024 •

edited

Loading